package bupt.xixi.registry.impl;

import bupt.xixi.loadbalancer.LoaderBalancer;
import bupt.xixi.loadbalancer.LoaderBalancerFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class ZkServiceRegistry extends AbstractRegistry {

    private static final String ZK_REGISTER_ROOT_PATH = "/my-rpc";
    private static final int BASE_SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;
    // private static final BalancerType BALANCE = BalancerType.ROUND_ROBIN_BALANCER;

    // 记录注册过的服务子节点
    private final Set<String> CACHE_REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet(); // 记录所有注册过的服务
    private final Map<String, List<String>> CACHE_SERVICE_ADDRESS_MAP = new ConcurrentHashMap<>();
    private CuratorFramework zkClient;
    private LoaderBalancer balancer;


    @Override
    protected void doRegistry(String serviceName, InetSocketAddress inetSocketAddress) {
        if (zkClient == null) {
            connect();
        }

        // 1. 构建服务节点全路径名
        String path = ZK_REGISTER_ROOT_PATH + "/" + serviceName + "/" + inetSocketAddress.toString();
        try {
            // 2. 判断是否重复节点
            if (zkClient.checkExists().forPath(path) != null) {
                log.info("节点已经存在. 该节点路径为:{}", path); // todo: 报错
            } else {
                // 3. 创建临时节点
                // eg: /my-rpc/rpc.serviceImplement.name/127.0.0.1:9999
                // 只是指定了path, 内容为空
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
                log.info("成功创建了节点: {}", path);
            }
            // 4. 为serviceName注册监听器, 同时更新map记录
            if (!CACHE_SERVICE_ADDRESS_MAP.containsKey(serviceName)) {
                registerWatcher(serviceName);
                CACHE_SERVICE_ADDRESS_MAP.put(serviceName,
                                              zkClient.getChildren().forPath(ZK_REGISTER_ROOT_PATH + "/" + serviceName));
            }
            if (!CACHE_REGISTERED_PATH_SET.contains(path)) {
                CACHE_REGISTERED_PATH_SET.add(path);
            }
        } catch (Exception e) {
            log.error("注册服务失败: {}", path);
        }
    }

    @Override
    protected void doShutdownHook() {
        log.info("注销钩子启动");
        CACHE_REGISTERED_PATH_SET.forEach(path -> {
            try {
                log.info("尝试注销zk节点ing -> [{}]", path);
                zkClient.delete().deletingChildrenIfNeeded().forPath(path);
            } catch (Exception e) {
                log.error("结束时注销zk服务失败: [{}]", e.getMessage());
            }
        });
    }

    private void connect() {
        LoaderBalancerFactory loaderBalancerFactory = new LoaderBalancerFactory();
        // balancer = loaderBalancerFactory.getBalancer(BALANCE.getCode());
        // 设置默认重连策略, 重连三次&每次间隔1s
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(retryPolicy).build();
        zkClient.start();
    }

    private void registerWatcher(String rpcServiceName) throws Exception {
        // 得到该父节点地址
        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName;
        // 开辟一块缓存记录该父节点下子节点的数据
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        // 创建监听器监测缓存变化: 通过lambda表达式实现单函数接口, 设置回调函数
        PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, pathChildrenCacheEvent) -> {
            // 只要子节点发生变化: 增删改等
            // 就更新内存中的服务列表!
            List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
            CACHE_SERVICE_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);

        };
        // 注册监听器
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        // 开启监听器
        pathChildrenCache.start();
    }
}
